/*-------------------------------------------------------------------------
 *
 * rangetypes_typanalyze.c
 *	  收集范围列统计信息的函数
 *
 * 对于范围类型列，收集下界和上界的直方图，以及
 * NULL和空范围的比例。
 *
 * 两个直方图具有相同的长度，并且它们被组合成一个
 * 单一的范围数组。这个数组的形状与
 * std_typanalyze收集的直方图相同，但值是不同的。
 * 数组中的每个范围都是一个有效范围，即使下界和
 * 上界来自不同的元组。理论上，标准的标量选择性
 * 函数也可以与组合直方图一起使用。
 *
 * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 *
 * IDENTIFICATION
 *	  src/backend/utils/adt/rangetypes_typanalyze.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include "catalog/pg_operator.h"
#include "commands/vacuum.h"
#include "utils/float.h"
#include "utils/fmgrprotos.h"
#include "utils/lsyscache.h"
#include "utils/rangetypes.h"
#include "utils/multirangetypes.h"

static int	fc_float8_qsort_cmp(const void *fc_a1, const void *fc_a2, void *fc_arg);
static int	fc_range_bound_qsort_cmp(const void *fc_a1, const void *fc_a2, void *fc_arg);
static void fc_compute_range_stats(VacAttrStats *fc_stats,
								AnalyzeAttrFetchFunc fc_fetchfunc, int fc_samplerows,
								double fc_totalrows);

/*
 * range_typanalyze -- 范围列的 typanalyze 函数
 */
Datum range_typanalyze(PG_FUNCTION_ARGS)
{
	VacAttrStats *fc_stats = (VacAttrStats *) PG_GETARG_POINTER(0);
	TypeCacheEntry *fc_typcache;
	Form_pg_attribute fc_attr = fc_stats->attr;

	/* 获取有关范围类型的信息；注意列可能是域 */
	fc_typcache = range_get_typcache(fcinfo, getBaseType(fc_stats->attrtypid));

	if (fc_attr->attstattarget < 0)
		fc_attr->attstattarget = default_statistics_target;

	fc_stats->compute_stats = fc_compute_range_stats;
	fc_stats->extra_data = fc_typcache;
	/* 与 std_typanalyze 相同 */
	fc_stats->minrows = 300 * fc_attr->attstattarget;

	PG_RETURN_BOOL(true);
}

/*
 * multirange_typanalyze -- 多范围列的 typanalyze 函数
 *
 * 我们对最小范围进行相同的分析，该范围完全包含多范围。
 */
Datum multirange_typanalyze(PG_FUNCTION_ARGS)
{
	VacAttrStats *fc_stats = (VacAttrStats *) PG_GETARG_POINTER(0);
	TypeCacheEntry *fc_typcache;
	Form_pg_attribute fc_attr = fc_stats->attr;

	/* 获取有关多范围类型的信息；注意列可能是域 */
	fc_typcache = multirange_get_typcache(fcinfo, getBaseType(fc_stats->attrtypid));

	if (fc_attr->attstattarget < 0)
		fc_attr->attstattarget = default_statistics_target;

	fc_stats->compute_stats = fc_compute_range_stats;
	fc_stats->extra_data = fc_typcache;
	/* 与 std_typanalyze 相同 */
	fc_stats->minrows = 300 * fc_attr->attstattarget;

	PG_RETURN_BOOL(true);
}

/*
 * 用于排序 float8 的比较函数，适用于范围长度。
 */
static int fc_float8_qsort_cmp(const void *fc_a1, const void *fc_a2, void *fc_arg)
{
	const float8 *fc_f1 = (const float8 *) fc_a1;
	const float8 *fc_f2 = (const float8 *) fc_a2;

	if (*fc_f1 < *fc_f2)
		return -1;
	else if (*fc_f1 == *fc_f2)
		return 0;
	else
		return 1;
}

/*
 * 排序 RangeBounds 的比较函数。
 */
static int fc_range_bound_qsort_cmp(const void *fc_a1, const void *fc_a2, void *fc_arg)
{
	RangeBound *fc_b1 = (RangeBound *) fc_a1;
	RangeBound *fc_b2 = (RangeBound *) fc_a2;
	TypeCacheEntry *fc_typcache = (TypeCacheEntry *) fc_arg;

	return range_cmp_bounds(fc_typcache, fc_b1, fc_b2);
}

/*
 * compute_range_stats() -- 计算范围列的统计信息
 */
static void fc_compute_range_stats(VacAttrStats *fc_stats, AnalyzeAttrFetchFunc fc_fetchfunc,
					int fc_samplerows, double fc_totalrows)
{
	TypeCacheEntry *fc_typcache = (TypeCacheEntry *) fc_stats->extra_data;
	TypeCacheEntry *fc_mltrng_typcache = NULL;
	bool		fc_has_subdiff;
	int			fc_null_cnt = 0;
	int			fc_non_null_cnt = 0;
	int			fc_non_empty_cnt = 0;
	int			fc_empty_cnt = 0;
	int			fc_range_no;
	int			fc_slot_idx;
	int			fc_num_bins = fc_stats->attr->attstattarget;
	int			fc_num_hist;
	float8	   *fc_lengths;
	RangeBound *fc_lowers,
			   *fc_uppers;
	double		fc_total_width = 0;

	if (fc_typcache->typtype == TYPTYPE_MULTIRANGE)
	{
		fc_mltrng_typcache = fc_typcache;
		fc_typcache = fc_typcache->rngtype;
	}
	else
		Assert(fc_typcache->typtype == TYPTYPE_RANGE);
	fc_has_subdiff = OidIsValid(fc_typcache->rng_subdiff_finfo.fn_oid);

	/* 分配内存以保存范围边界和样本范围的长度。 */
	fc_lowers = (RangeBound *) palloc(sizeof(RangeBound) * fc_samplerows);
	fc_uppers = (RangeBound *) palloc(sizeof(RangeBound) * fc_samplerows);
	fc_lengths = (float8 *) palloc(sizeof(float8) * fc_samplerows);

	/* 遍历样本范围。 */
	for (fc_range_no = 0; fc_range_no < fc_samplerows; fc_range_no++)
	{
		Datum		fc_value;
		bool		fc_isnull,
					fc_empty;
		MultirangeType *fc_multirange;
		RangeType  *fc_range;
		RangeBound	fc_lower,
					fc_upper;
		float8		fc_length;

		vacuum_delay_point();

		fc_value = fc_fetchfunc(fc_stats, fc_range_no, &fc_isnull);
		if (fc_isnull)
		{
			/* 范围为null，仅计算该值 */
			fc_null_cnt++;
			continue;
		}

		/*
		 * XXX: 我们是否应该忽略宽值，例如std_typanalyze，以
		 * 避免膨胀统计表？
		 */
		fc_total_width += VARSIZE_ANY(DatumGetPointer(fc_value));

		/* 获取范围并反序列化以便进一步分析。 */
		if (fc_mltrng_typcache != NULL)
		{
			/* 将多范围视为一个没有间隔的大范围。 */
			fc_multirange = DatumGetMultirangeTypeP(fc_value);
			if (!MultirangeIsEmpty(fc_multirange))
			{
				RangeBound	fc_tmp;

				multirange_get_bounds(fc_typcache, fc_multirange, 0,
									  &fc_lower, &fc_tmp);
				multirange_get_bounds(fc_typcache, fc_multirange,
									  fc_multirange->rangeCount - 1,
									  &fc_tmp, &fc_upper);
				fc_empty = false;
			}
			else
			{
				fc_empty = true;
			}
		}
		else
		{
			fc_range = DatumGetRangeTypeP(fc_value);
			range_deserialize(fc_typcache, fc_range, &fc_lower, &fc_upper, &fc_empty);
		}

		if (!fc_empty)
		{
			/* 记住边界和长度，以在直方图中进一步使用 */
			fc_lowers[fc_non_empty_cnt] = fc_lower;
			fc_uppers[fc_non_empty_cnt] = fc_upper;

			if (fc_lower.infinite || fc_upper.infinite)
			{
				/* 任何类型的无限范围的长度都是无限的 */
				fc_length = get_float8_infinity();
			}
			else if (fc_has_subdiff)
			{
				/*/
				 * 对于普通范围，使用上界和下界值之间的 subdiff 函数。
				 */
				fc_length = DatumGetFloat8(FunctionCall2Coll(&fc_typcache->rng_subdiff_finfo,
														  fc_typcache->rng_collation,
														  fc_upper.val, fc_lower.val));
			}
			else
			{
				/* 如果没有可用的 subdiff，则使用默认值 1.0。 */
				fc_length = 1.0;
			}
			fc_lengths[fc_non_empty_cnt] = fc_length;

			fc_non_empty_cnt++;
		}
		else
			fc_empty_cnt++;

		fc_non_null_cnt++;
	}

	fc_slot_idx = 0;

	/* 只有在找到一些非空值的情况下，我们才能计算真实的统计数据。 */
	if (fc_non_null_cnt > 0)
	{
		Datum	   *fc_bound_hist_values;
		Datum	   *fc_length_hist_values;
		int			fc_pos,
					fc_posfrac,
					fc_delta,
					fc_deltafrac,
					fc_i;
		MemoryContext fc_old_cxt;
		float4	   *fc_emptyfrac;

		fc_stats->stats_valid = true;
		/* 进行简单的空值比例和宽度统计 */
		fc_stats->stanullfrac = (double) fc_null_cnt / (double) fc_samplerows;
		fc_stats->stawidth = fc_total_width / (double) fc_non_null_cnt;

		/* 估计非空值是唯一的 */
		fc_stats->stadistinct = -1.0 * (1.0 - fc_stats->stanullfrac);

		/* 必须将目标值复制到anl_context中 */
		fc_old_cxt = MemoryContextSwitchTo(fc_stats->anl_context);

		/*/
		 * 如果至少有两个值，则生成一个边界直方图槽条目。
		 */
		if (fc_non_empty_cnt >= 2)
		{
			/* 排序边界值 */
			qsort_interruptible(fc_lowers, fc_non_empty_cnt, sizeof(RangeBound),
								fc_range_bound_qsort_cmp, fc_typcache);
			qsort_interruptible(fc_uppers, fc_non_empty_cnt, sizeof(RangeBound),
								fc_range_bound_qsort_cmp, fc_typcache);

			fc_num_hist = fc_non_empty_cnt;
			if (fc_num_hist > fc_num_bins)
				fc_num_hist = fc_num_bins + 1;

			fc_bound_hist_values = (Datum *) palloc(fc_num_hist * sizeof(Datum));

			/*/
			 * 这个循环的目的是根据 lowers[] 和 uppers[] 中的第一个和最后一个条目构建范围，以及之间均匀间隔的值。因此，第 i 个值是 lowers[(i * (nvals - 1)) / (num_hist - 1)] 和 uppers[(i * (nvals - 1)) / (num_hist - 1)] 的范围。但是直接计算该下标存在整数溢出的风险，当统计目标超过几千时。相反，我们在每一步将 (nvals - 1) / (num_hist - 1) 加到 pos，分别跟踪和的整数和小数部分。 */
			fc_delta = (fc_non_empty_cnt - 1) / (fc_num_hist - 1);
			fc_deltafrac = (fc_non_empty_cnt - 1) % (fc_num_hist - 1);
			fc_pos = fc_posfrac = 0;

			for (fc_i = 0; fc_i < fc_num_hist; fc_i++)
			{
				fc_bound_hist_values[fc_i] = PointerGetDatum(range_serialize(fc_typcache,
																	   &fc_lowers[fc_pos],
																	   &fc_uppers[fc_pos],
																	   false));
				fc_pos += fc_delta;
				fc_posfrac += fc_deltafrac;
				if (fc_posfrac >= (fc_num_hist - 1))
				{
					/* 小数部分超过 1，将其进位到整数部分 */
					fc_pos++;
					fc_posfrac -= (fc_num_hist - 1);
				}
			}

			fc_stats->stakind[fc_slot_idx] = STATISTIC_KIND_BOUNDS_HISTOGRAM;
			fc_stats->stavalues[fc_slot_idx] = fc_bound_hist_values;
			fc_stats->numvalues[fc_slot_idx] = fc_num_hist;

			/* 即使我们正在分析一个多范围列，也要存储范围 */
			fc_stats->statypid[fc_slot_idx] = fc_typcache->type_id;
			fc_stats->statyplen[fc_slot_idx] = fc_typcache->typlen;
			fc_stats->statypbyval[fc_slot_idx] = fc_typcache->typbyval;
			fc_stats->statypalign[fc_slot_idx] = fc_typcache->typalign;

			fc_slot_idx++;
		}

		/*/
		 * 如果至少有两个值，则生成一个长度直方图槽条目。
		 */
		if (fc_non_empty_cnt >= 2)
		{
			/*/
			 * 上升排序范围长度以进一步填充直方图 */
			qsort_interruptible(fc_lengths, fc_non_empty_cnt, sizeof(float8),
								fc_float8_qsort_cmp, NULL);

			fc_num_hist = fc_non_empty_cnt;
			if (fc_num_hist > fc_num_bins)
				fc_num_hist = fc_num_bins + 1;

			fc_length_hist_values = (Datum *) palloc(fc_num_hist * sizeof(Datum));

			/*/
			 * 该循环的目的是复制 lengths[] 中的第一个和最后一个条目，以及之间均匀间隔的值。因此，第 i 个值是 lengths[(i * (nvals - 1)) / (num_hist - 1)]。但是直接计算该下标存在整数溢出的风险，当统计目标超过几千时。相反，我们在每一步将 (nvals - 1) / (num_hist - 1) 加到 pos，分别跟踪和的整数和小数部分。 */
			fc_delta = (fc_non_empty_cnt - 1) / (fc_num_hist - 1);
			fc_deltafrac = (fc_non_empty_cnt - 1) % (fc_num_hist - 1);
			fc_pos = fc_posfrac = 0;

			for (fc_i = 0; fc_i < fc_num_hist; fc_i++)
			{
				fc_length_hist_values[fc_i] = Float8GetDatum(fc_lengths[fc_pos]);
				fc_pos += fc_delta;
				fc_posfrac += fc_deltafrac;
				if (fc_posfrac >= (fc_num_hist - 1))
				{
					/* 小数部分超过 1，将其进位到整数部分 */
					fc_pos++;
					fc_posfrac -= (fc_num_hist - 1);
				}
			}
		}
		else
		{
			/*/
			 * 即使我们不创建直方图，也要存储一个空数组以表示“没有直方图”。我们不能仅仅把 stavalues 留为 NULL，因为如果请求 stavalues 时其为 NULL，get_attstatsslot() 会报错。我们仍会将空分数存储在 stanumbers 中。 */
			fc_length_hist_values = palloc(0);
			fc_num_hist = 0;
		}
		fc_stats->staop[fc_slot_idx] = Float8LessOperator;
		fc_stats->stacoll[fc_slot_idx] = InvalidOid;
		fc_stats->stavalues[fc_slot_idx] = fc_length_hist_values;
		fc_stats->numvalues[fc_slot_idx] = fc_num_hist;
		fc_stats->statypid[fc_slot_idx] = FLOAT8OID;
		fc_stats->statyplen[fc_slot_idx] = sizeof(float8);
		fc_stats->statypbyval[fc_slot_idx] = FLOAT8PASSBYVAL;
		fc_stats->statypalign[fc_slot_idx] = 'd';

		/* 存储空范围的分数 */
		fc_emptyfrac = (float4 *) palloc(sizeof(float4));
		*fc_emptyfrac = ((double) fc_empty_cnt) / ((double) fc_non_null_cnt);
		fc_stats->stanumbers[fc_slot_idx] = fc_emptyfrac;
		fc_stats->numnumbers[fc_slot_idx] = 1;

		fc_stats->stakind[fc_slot_idx] = STATISTIC_KIND_RANGE_LENGTH_HISTOGRAM;
		fc_slot_idx++;

		MemoryContextSwitchTo(fc_old_cxt);
	}
	else if (fc_null_cnt > 0)
	{
		/* 我们只找到空值；假设该列完全为空 */
		fc_stats->stats_valid = true;
		fc_stats->stanullfrac = 1.0;
		fc_stats->stawidth = 0;	/* "unknown" */
		fc_stats->stadistinct = 0.0;	/* "unknown" */
	}

	/*
	 * 我们不需要清理任何临时的 palloc。哈希表也应该被清除，因为它使用了子内存上下文。
	 */
}
